package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Demo07Car {
    public static void main(String[] args) throws Exception {
        /*
         * 需求：基于卡口过车数据统计道路的拥堵情况
         * 拥堵情况：通过车速以及车流量进行判断
         * 思路：通过对道路以及卡口进行分组，统计车流量以及平均车速
         * 写代码的思路：
         *      1、接入数据，通过Socket模拟实时过车数据
         *      2、提取数据中的时间，并设置事件时间及水位线
         *      3、按照道路及卡口分组，使用滑动窗口每隔1分钟统计最近10分钟内车流量信息
         *      4、打印数据
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> lineDS = env.socketTextStream("master", 8888);

        // 对每条数据进行切分，并将其转换成Car的对象
        SingleOutputStreamOperator<Car> carDS = lineDS.map(line -> {
            String[] splits = line.split(",");
            return new Car(splits[0], Integer.parseInt(splits[1]), Integer.parseInt(splits[2])
                    , splits[3], splits[4], splits[5]
                    , splits[6], Long.parseLong(splits[7]) * 1000, Double.parseDouble(splits[8])
            );
        });

        // 设置事件时间以及水位线
        SingleOutputStreamOperator<Car> assignDS = carDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Car>forMonotonousTimestamps()
                        .withTimestampAssigner((car, ts) -> car.getTime())
        );

        // 按照卡口以及道路id进行分组
        assignDS.keyBy(car -> car.getRoadId() + "|" + car.getkId())
                // 使用滑动窗口，每隔1分钟统计最近10分钟内的卡口数据
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
                // 统计车流量又需要统计平均车速
                // 没有直接的API可以使用，需要使用底层的API自定义窗口的计算逻辑
                .process(new ProcessWindowFunction<Car, Result, String, TimeWindow>() {
                    /**
                     *
                     * @param s 分组的Key，由卡口编号及道路编号拼接而成
                     * @param context Flink任务运行时的上下文环境
                     * @param elements 每个分组每个窗口接收到的数据，即最近10分钟内的数据
                     * @param out 用于将结果输出到下游
                     */
                    // 每个窗口会执行一次，即1分钟执行一次
                    @Override
                    public void process(String s, ProcessWindowFunction<Car, Result, String, TimeWindow>.Context context, Iterable<Car> elements, Collector<Result> out) throws Exception {
                        // 提取道路id以卡口id
                        String[] roadAndKId = s.split("\\|");
                        int cnt = 0;
                        double sumSpeed = 0;
                        for (Car car : elements) {
                            sumSpeed += car.getSpeed();
                            // 统计车流量
                            cnt++;
                        }
                        // 计算平均车速
                        double avgSpeed = sumSpeed / cnt;
                        // 构建Result对象，通过out进行输出
                        out.collect(new Result(roadAndKId[0], roadAndKId[1], cnt, avgSpeed));
                    }
                }).print();

        env.execute();

    }
}

class Result {
    private String roadId;
    private String kId;
    private Integer carCnt;
    private Double avgSpeed;

    public Result(String roadId, String kId, Integer carCnt, Double avgSpeed) {
        this.roadId = roadId;
        this.kId = kId;
        this.carCnt = carCnt;
        this.avgSpeed = avgSpeed;
    }

    @Override
    public String toString() {
        return "Result{" +
                "roadId='" + roadId + '\'' +
                ", kId='" + kId + '\'' +
                ", carCnt=" + carCnt +
                ", avgSpeed=" + avgSpeed +
                '}';
    }
}


